package reactor;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.CharsetUtil;
import com.google.common.io.Files;
import lombok.SneakyThrows;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.Closeable;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.charset.Charset;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.stream.Stream;

/**
 * using
 *
 * @author qingyan 2022-01-20
 */
public class ReactorUsingTest {
    /**
     *
     *
     *
     */
    @Test
    public void fileReads(){
        File file = FileUtil.file("C:\\Users\\Administrator\\Desktop\\temp\\万科集团OA20180305.txt");
        List<String> list = FileUtil.readLines(file, CharsetUtil.UTF_8);
    }
    @Test
    public void reads(){
        File file = new File("C:\\Users\\Administrator\\Desktop\\temp\\万科集团OA20180305.txt");
        BufferedReader reader = null;
        try {
            System.out.println("以行为单位读取文件内容，一次读一整行：");
            reader = new BufferedReader(new FileReader(file),8000);
            String tempString = null;
            int line = 1;
            // 一次读入一行，直到读入null为文件结束
            while ((tempString = reader.readLine()) != null) {
                // 显示行号
//                System.out.println("line " + line + ": " + tempString);
                line++;
            }
            reader.close();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e1) {
                }
            }
        }

    }

    @Test
    public void nioRead(){
        File file = new File("C:\\Users\\Administrator\\Desktop\\temp\\万科集团OA20180305.txt");
        try {
            List<String> list = Files.readLines(file, CharsetUtil.defaultCharset());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @SneakyThrows
    @Test
    public void usingTest(){
        File file = new File("C:\\Users\\Administrator\\Desktop\\temp\\万科集团OA20180305.txt");
        CountDownLatch latch = new CountDownLatch(3);
        Flux<String> stringFlux = Flux.using(
                readLines(file),
                Flux::fromStream,
                s -> System.out.println("关闭！")
        ).subscribeOn(Schedulers.newParallel("file-copy", 3))
                .share();
//        File copy = new File("C:\\Users\\Administrator\\Desktop\\temp\\COPY万科集团OA20180305.txt");
        // output file
        BufferedWriter bw1 = Files.newWriter(new File("C:\\Users\\Administrator\\Desktop\\temp\\COPY1万科集团OA20180305.txt"), CharsetUtil.defaultCharset());
        BufferedWriter bw2 = Files.newWriter(new File("C:\\Users\\Administrator\\Desktop\\temp\\COPY2万科集团OA20180305.txt"), CharsetUtil.defaultCharset());
        BufferedWriter bw3 = Files.newWriter(new File("C:\\Users\\Administrator\\Desktop\\temp\\COPY3万科集团OA20180305.txt"), CharsetUtil.defaultCharset());
        // close file if error / oncomplete
        stringFlux
                .subscribe(s -> write(bw1, s),
                        (e) -> close(bw1),
                        () -> {
                            close(bw1);
                            latch.countDown();
                        }
                );
        stringFlux
                .subscribe(s -> write(bw2, s),
                        (e) -> close(bw2),
                        () -> {
                            close(bw2);
                            latch.countDown();
                        }
                );
        stringFlux
                .subscribe(s -> write(bw3, s),
                        (e) -> close(bw3),
                        () -> {
                            close(bw3);
                            latch.countDown();
                        }
                );

        latch.await();

    }

    public Callable<Stream<String>> readLines(File file) {
        return () -> {
            return Files.readLines(file, CharsetUtil.defaultCharset()).stream();
        };
    }

    // private methods to handle checked exceptions

    private void close(Closeable closeable){
        try {
            closeable.close();
            System.out.println("Closed the resource");
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    private void write(BufferedWriter bw, String string){
        try {
            bw.write(string);
            bw.newLine();
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }




}
